package org.wenzi.com.job;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.io.PojoCsvInputFormat;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.wenzi.com.agg.CountAgg;
import org.wenzi.com.function.TopNFunction;
import org.wenzi.com.function.WindowResultFunction;
import org.wenzi.com.pojo.UserBehaviorPojo;

import java.io.File;
import java.net.URL;

/**
 * @author zhaozuowen
 * @date 2021-10-25 10:52
 */
public class HotItemsJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // UserBehavior.csv 的本地文件路径
        URL fileUrl = HotItemsJob.class.getClassLoader().getResource("UserBehavior.csv");
        Path filePath = Path.fromLocalFile(new File(fileUrl.toURI()));
       // 抽取 UserBehavior 的 TypeInformation，是一个 PojoTypeInfo
        PojoTypeInfo<UserBehaviorPojo> pojoType = (PojoTypeInfo<UserBehaviorPojo>) TypeExtractor.createTypeInfo(UserBehaviorPojo.class);
        // 由于 Java 反射抽取出的字段顺序是不确定的，需要显式指定下文件中字段的顺序
        String[] fieldOrder = new String[]{"userId", "itemId", "categoryId", "behavior", "timestamp"};
       // 创建 PojoCsvInputFormat
        PojoCsvInputFormat<UserBehaviorPojo> csvInput = new PojoCsvInputFormat<>(filePath, pojoType, fieldOrder);


        DataStreamSource<UserBehaviorPojo> input = env.createInput(csvInput,pojoType);


        SingleOutputStreamOperator<UserBehaviorPojo> timeData = input.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehaviorPojo>() {
            @Override
            public long extractAscendingTimestamp(UserBehaviorPojo o) {
                return o.getTimestamp() * 1000;
            }
        });
        SingleOutputStreamOperator<UserBehaviorPojo> pvData = timeData.filter((FilterFunction<UserBehaviorPojo>) userBehaviorPojo -> userBehaviorPojo.getBehavior().equalsIgnoreCase("pv"));
        SingleOutputStreamOperator<WindowResultFunction.ItemViewCount> windowData = pvData.keyBy("itemId").timeWindow(
                Time.minutes(60), Time.minutes(5)).aggregate(new CountAgg(), new WindowResultFunction());

        SingleOutputStreamOperator<String> topNStream = windowData.keyBy("windowEnd").process(new TopNFunction(3));
        topNStream.print();
        env.execute("HotItems job execute");


    }
}
